#include"subscribe.h"

extern tcp_server *server;

// 指向当前单例对象的指针
SubscribeList* SubscribeList::_instance = NULL;
//用于保证创建单例的init方法只执行一次的锁
pthread_once_t SubscribeList::_once = PTHREAD_ONCE_INIT;

// 构造函数
SubscribeList::SubscribeList(){
    // 将订阅清单和发布清单的锁初始化
    pthread_mutex_init(&_book_list_lock, NULL);
    pthread_mutex_init(&_push_list_lock, NULL);


}



// 订阅功能
void SubscribeList::subscribe(uint64_t mod, int fd){
    // 加锁
    pthread_mutex_lock(&_book_list_lock);
    // 将数据放入map中
    _book_list[mod].insert(fd);
    // 解锁
    pthread_mutex_unlock(&_book_list_lock);
}

// 取消订阅功能
void SubscribeList::unsubscribe(uint64_t mod, int fd){
    int modid = (int)(mod>>32);
    int cmdid = (int)(mod);
    
    // 加锁
    pthread_mutex_lock(&_book_list_lock);
    
    if(_book_list.find(mod) != _book_list.end()){
        // 将数据从_book_list map中去除
        _book_list[mod].erase(fd);
        printf("fd:[%d] unsubscribe modid:[%d], cmdid:[%d]\n",fd, modid, cmdid);
        // 判断去除后该mod是否还有数据
        if(_book_list[mod].empty()){
            _book_list.erase(mod);
            printf("_book_list[%lu] is empty!\n",mod);
        }

        // 将数据从_push_list map中去除
        _push_list[fd].erase(mod);
        // 判断去除后该mod是否还有数据
        if(_push_list[fd].empty()){
            _push_list.erase(fd);
            printf("_push_list[%d] is empty!\n", fd);
        }

    }
    else{
        printf("The map does not have this element\n");
    }

    // 解锁
    pthread_mutex_unlock(&_book_list_lock);
}

// 主动推送任务
void push_change_task(event_loop* loop, void* args){
    SubscribeList* subscribe = (SubscribeList*)args;
    
    // 1.得到目前在线的fd都有哪些
    listen_fd_set online_fds;
    loop->get_fds(online_fds);

    // 2.从_push_list中找到与online_fds集合匹配的fd,放在一个新的publish_map里
    publish_map need_publish;  // 真正需要发布的订单
    subscribe->make_publish_map(online_fds, need_publish);
    // 3.依次从need_publish中取出数据，发送给对应的客户端
    publish_map::iterator it;
    for(it=need_publish.begin(); it!=need_publish.end(); it++){
        // 4.获取在线的fd
        int fd = it->first;
        
        // 5.遍历当前fd下的mod
        __gnu_cxx::hash_set<uint64_t>::iterator sit;
        for(sit=it->second.begin(); sit!=it->second.end(); sit++){
            // 6.获取modid和cmdid消息
            int modId = (int)((*sit)>>32);
            int cmdId = (int)(*sit);
            
            // 7.组装proto消息
            lars::GetRouteResponse response;
            response.set_modid(modId);
            response.set_cmdid(cmdId);

            // 8.通过route查询对应的主机的Ip和Port进行组装
            host_set hosts = Route::instance()->get_hosts(modId, cmdId);

            // 9.遍历所有的主机
            for(host_set_it hit=hosts.begin(); hit!=hosts.end(); hit++){
                // 此时的hit就是ip+port组成的uint64_t数据
                lars::HostInfo host;
                host.set_ip((uint32_t)((*hit)>>32));
                host.set_port((uint32_t)(*hit));
                
                response.add_host()->CopyFrom(host);
            }

            // 10.将response转换成字符串
            std::string responseString;
            response.SerializeToString(&responseString);

            // 11.将打包好的proto数据通过fd取出链接发送回去
            net_connection *conn = tcp_server::conns[fd];
            if (conn != NULL) {
                conn->send_message(responseString.c_str(), responseString.size(), lars::ID_GetRouteResponse);
            }
            else{
                fprintf(stderr, "publish conn is NULL1\n");
            }

            // 将数据发送出去后在删除_push_list中的数据
            SubscribeList::instance()->get_push_list().erase(fd);
        }
    }
}

// 发布功能
void SubscribeList::publish(std::vector<uint64_t> &change_mods){
    // 加锁
    pthread_mutex_lock(&_book_list_lock);
    pthread_mutex_lock(&_push_list_lock);
    // 遍历其中的每一个mod
    std::vector<uint64_t>::iterator it;
    for(it=change_mods.begin(); it!=change_mods.end(); it++){
        uint64_t mod = *it;
        // 查找目前是否有用户订阅了该mod
        if(_book_list.find(mod) != _book_list.end()){
            // 当前被修改的mod在订阅清单中
            // 将当前订阅了当前mod的fd加入到_push_list中
            __gnu_cxx::hash_set<int>::iterator fds_it;
            for(fds_it=_book_list[mod].begin(); fds_it!=_book_list[mod].end(); fds_it++){
                int fd = *fds_it;
                _push_list[fd].insert(mod);
            }
        }
    }
    // 解锁，先给谁加锁则后给谁解锁，加锁和解锁的顺序相反
    pthread_mutex_unlock(&_push_list_lock);
    pthread_mutex_unlock(&_book_list_lock);
    // 通知server，给各个线程去执行推送任务,告知所有线程去通知给_push_list中的每一个fd发送新的mod
    server->set_task(push_change_task, this);


}


//根据在线用户fd得到真正需要发布的列表
void SubscribeList::make_publish_map(listen_fd_set &online_fds, publish_map &need_publish){
    // 遍历_push_list，找到online_fds中存在的fd，将其数据放入到need_publish中
    publish_map::iterator it;

    // 加锁
    pthread_mutex_lock(&_push_list_lock);

    for(it=_push_list.begin(); it!=_push_list.end(); it++){
        // printf("1_push_list fd:[%d]\n", it->first);
        if(online_fds.find(it->first) != online_fds.end()){
            // 表明当前fd在线，将当前fd的_push_list加入到need_publish中
            // 浅拷贝，当it删除后会出问题
            // need_publish[it->first] = _push_list[it->first];   // 两个set的赋值

            // 使用深拷贝
            __gnu_cxx::hash_set<uint64_t>::iterator hit;
            for(hit=_push_list[it->first].begin(); hit!=_push_list[it->first].end(); hit++){
                need_publish[it->first].insert(*hit);
            }

            // 单线程是不会出问题，多线程则会删除出错
            // _push_list.erase(it); // 删除当前元素
            
        }
        
        
        // printf("2_push_list fd:[%d]\n", it->first);
    }
    // 解锁
    pthread_mutex_unlock(&_push_list_lock);

}
